package org.example.netty.gRPCTest;

import com.shengsiyuan.proto.*;
import io.grpc.ManagedChannel;
import io.grpc.ManagedChannelBuilder;
import io.grpc.stub.StreamObserver;

public class GrpcClient {
    public static void main(String[] args) {
//        new GrpcClient().ServerStreamRPC();
//        new GrpcClient().ClientStreamRPC();
        new GrpcClient().BiStreamRPC();
    }
    // 简单RPC调用
     void SimpleRPC(){
        // 创建客户端, 指定服务端地址和端口
        ManagedChannel channel = ManagedChannelBuilder.forAddress("localhost", 50051)
                .usePlaintext()
                .build();
        // 创建服务端stub, 绑定channel
        StudentServiceGrpc.StudentServiceBlockingStub stub = StudentServiceGrpc.newBlockingStub(channel);
        MyRequest p = MyRequest.newBuilder().setUsername("ak").build();
        MyResponse res = stub.getRealNameByUsername(p);
        System.out.println(res.getRealname());
    }
    // 服务端流式RPC调用
    void ServerStreamRPC(){
        // 创建客户端, 指定服务端地址和端口
        ManagedChannel channel = ManagedChannelBuilder.forAddress("localhost", 50051)
                .usePlaintext()
                .build();
        // 创建服务端stub, 绑定channel
        StudentServiceGrpc.StudentServiceBlockingStub stub = StudentServiceGrpc.newBlockingStub(channel);
        StudentRequestAge p = StudentRequestAge.newBuilder().setAge(19).build();
        java.util.Iterator<com.shengsiyuan.proto.StudentResponse> iterator = stub.getStudentsByAge(p);
        while (iterator.hasNext()){
            System.out.println(iterator.next());
        }
    }
    // 客户端流式RPC调用
    void ClientStreamRPC(){
        // 创建客户端, 指定服务端地址和端口
        ManagedChannel channel = ManagedChannelBuilder.forAddress("localhost", 50051)
                .usePlaintext()
                .build();
        // 创建服务端stub, 绑定channel,同步阻塞
//        StudentServiceGrpc.StudentServiceBlockingStub stub = StudentServiceGrpc.newBlockingStub(channel);
        //  创建客户端流式RPC的回调函数, 用于处理服务端返回的数据
        StreamObserver<StudentResponseList> studentResponseListStreamObserver = new StreamObserver<StudentResponseList>(){
            @Override
            public void onNext(StudentResponseList studentResponseList) {
                for (StudentResponse studentResponse : studentResponseList.getStudentList()){
                    System.out.println(studentResponse);
                }
            }

            @Override
            public void onError(Throwable throwable) {

            }
            // 回调函数结束, 用于处理服务端返回的最终结果, 比如返回是否成功等
            @Override
            public void onCompleted() {

            }
        };
        //客户端使用流式向服务端发送数据，都是异步的
        //StudentServiceGrpc.StudentServiceBlockingStub是同步阻塞的
        StudentServiceGrpc.StudentServiceStub stub = StudentServiceGrpc.newStub(channel);
        //异步发送数据
        StreamObserver<StudentRequestAge> studentRequestAgeStreamObserver = stub.getStudentsByAges(studentResponseListStreamObserver);
        for (int i=0;i<10;i++){
            StudentRequestAge p = StudentRequestAge.newBuilder().setAge(20+i).build();
            studentRequestAgeStreamObserver.onNext(p);
        }
        studentRequestAgeStreamObserver.onCompleted();
        // 异步等待结果
        try {
            Thread.sleep(1000);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
    }
    // 双向流式RPC调用
    void BiStreamRPC(){
        // 创建客户端, 指定服务端地址和端口
        ManagedChannel channel = ManagedChannelBuilder.forAddress("localhost", 50051)
                .usePlaintext()
                .build();
        StudentServiceGrpc.StudentServiceStub stub = StudentServiceGrpc.newStub(channel);
        // 创建客户端流式RPC的回调函数, 用于处理服务端返回的数据
        StreamObserver<StreamRequest> streamRequestStreamObserver = stub.biTalk(new StreamObserver<StreamResponse>() {
            @Override
            public void onNext(StreamResponse streamResponse) {
                System.out.println("客户端收到服务端消息:" + streamResponse);
            }

            @Override
            public void onError(Throwable throwable) {

            }

            @Override
            public void onCompleted() {
                System.out.println("客户端收到服务端消息结束");
            }
        });
        // 异步发送数据
        for (int i=0;i<10;i++){
            StreamRequest p = StreamRequest.newBuilder().setRequestInfo("hello " + i).build();
            streamRequestStreamObserver.onNext(p);
        }
        streamRequestStreamObserver.onCompleted();

        // 异步等待结果
        try {
            Thread.sleep(1000);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
    }
}
